package com.lxl.core;

import com.lxl.config.TaskStatus;
import com.lxl.pojo.InterfaceMonitor;
import com.lxl.service.InterfaceMonitorService;
import org.apache.log4j.Logger;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransListener;
import org.pentaho.di.trans.TransMeta;

import java.util.Date;
import java.util.Iterator;
import java.util.Map;

public class EtlTransTask extends BaseTask {
    private static Logger logger = Logger.getLogger(EtlTransTask.class);

    public TaskResultMeta runTask() throws KettleException {
        TaskResultMeta resultMeta = new TaskResultMeta();

        //创建transformation
        EtlTransListener transListener = new EtlTransListener();
        TransMeta meta = new TransMeta(taskDataMeta.taskLocalConfigFilePath);
        Trans trans = new Trans(meta);

        //添加监听器
        transListener.setInterfaceMonitor(this.taskDataMeta.interfaceMonitor);
        trans.addTransListener(transListener);

        //传递参数到transformation
        Iterator<Map.Entry<String, String>> transParam = taskDataMeta.etlInputVariables.entrySet().iterator();
        while (transParam.hasNext()){
            Map.Entry<String, String> entry = transParam.next();
            trans.setVariable(entry.getKey(), entry.getValue());
        }

        //执行transformation
        trans.execute(null);
        trans.waitUntilFinished();
        //返回结果数据
        resultMeta.setCode(trans.getErrors());
        if (resultMeta.getCode() > 0){
            resultMeta.setResult(null);
        }else{
            resultMeta.setResult(trans.getVariable(taskDataMeta.etlResultKey));
        }
        return resultMeta;
    }

    class EtlTransListener implements TransListener{
        private InterfaceMonitor interfaceMonitor;
        private long start,end;
        private void setInterfaceMonitor(InterfaceMonitor interfaceMonitor){
            this.interfaceMonitor = interfaceMonitor;
        }

        private InterfaceMonitor getInterfaceMonitor(){
            return interfaceMonitor;
        }

        @Override
        public void transStarted(Trans trans) throws KettleException {
            interfaceMonitor.setStatus(TaskStatus.START);
            start = System.currentTimeMillis();
        }

        @Override
        public void transActive(Trans trans) {
        }

        @Override
        public void transFinished(Trans trans) throws KettleException {
            end = System.currentTimeMillis();
            interfaceMonitor.setStatus(TaskStatus.SUCCESS);
            interfaceMonitor.setTimeConsuming(end-start);
        }
    }
}
